package org.nbict.iot.task.realtime;

import com.alibaba.fastjson.JSONObject;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.mongodb.common.SimpleQueryFilterCreator;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.bson.Document;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.task.realtime.filter.AuthFilter;
import org.nbict.iot.task.realtime.filter.UploadFilter;
import org.nbict.iot.task.realtime.func.PacketJsonProject;
import org.nbict.iot.task.realtime.func.PacketResolver;
import org.nbict.iot.task.realtime.state.Comparer;
import org.nbict.iot.task.realtime.state.Timely;
import org.nbict.iot.task.realtime.state.TimelyFactory;
import org.nbict.iot.task.realtime.state.TimelyReplacer;
import org.nbict.iot.trident.kafka.KafkaSpout4ParsedTools;
import org.nbict.iot.trident.mogo.mapper.MongoMapper;
import org.nbict.iot.trident.mogo.mapper.impl.SimpleMongoMapper;
import org.nbict.iot.trident.mogo.state.MongoState;
import org.nbict.iot.trident.mogo.state.MongoStateFactory;
import org.nbict.iot.trident.mogo.state.MongoStateUpdater;

/**
 * Created by songseven on 18/6/29.
 */
public class RealtimeExactlyTrident {
    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        //设置kafka消费策略
        KafkaSpoutConfig.FirstPollOffsetStrategy strategy = KafkaSpoutConfig
                .FirstPollOffsetStrategy.EARLIEST;
        if (args.length > 1) {
            strategy = KafkaSpoutConfig.FirstPollOffsetStrategy.valueOf
                    (args[1]); //strategy name
        }

        String url = PropertyUtil.getProperty("mongo.url");
        String database = PropertyUtil.getProperty("mongo.database");
        String username = PropertyUtil.getProperty("mongo.username");
        String password = PropertyUtil.getProperty("mongo.password");
        String admin = PropertyUtil.getProperty("mongo.admin.name");
        String collectionName = "car";
        String collectionName1 = "car_authc";

        TridentTopology topology = new TridentTopology();

        KafkaSpoutConfig spoutConfig = KafkaSpout4ParsedTools.newKafkaSpoutConfig
                ("192.168.1.83:9092", "transit-.*-simple", // bootstrap server,
                        "iot-car-group-realtime-update", strategy); //group id ignored
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4ParsedTools
                .newKafkaTridentSpoutOpaque(spoutConfig);

        //重要，由流名称决定消费者组
        Stream stream = topology.newStream("iot-car-parsed-4-realtime", kafkaSpout);

        Fields packetFields = new Fields("car_packet_header",
                "car_packet_body");
        Fields outputFields = new Fields("car_vin", "action", "timestamp",
                "upload");

        Stream stage = stream.each(new Fields("platform_id", "packet"), new PacketJsonProject(packetFields), packetFields)
                .each(new Fields("platform_id", "car_packet_header", "car_packet_body"), new PacketResolver(), outputFields)
                .project(new Fields("platform_id", "car_vin", "action", "timestamp", "upload")); //去除packet元组

        Stream uploadStream = stage.each(outputFields, new UploadFilter()) //filter实时上报信息流
                .groupBy(new Fields("car_vin"))
                .toStream();

        //mongo数据库连接信息
        Timely.Options options = new Timely.Options()
                .withUrl(url)
                .withDatabase(database)
                .withUsername(username)
                .withPassword(password)
                .withAdmin(admin)
                .withCollectionName(collectionName);

        Fields persistFields = new Fields("platform_id", "car_vin", "timestamp", "upload");
        //"platform_id", "car_vin", "timestamp", "upload"
        uploadStream
                .partitionPersist(
                    new TimelyFactory(
                            options,
                            new Comparer() { //持久化
                                    @Override
                                    public Document compareAndSet(TridentTuple tuple, Document replaced) {
                                        Document document = new Document();
                                        document.append("platform_id", tuple.get(0)); //platform
                                        document.append("car_vin", tuple.get(1)); //vin
                                        Long timestamp = tuple.getLong(2); //timestamp
                                        JSONObject upload = (JSONObject) tuple.get(3);
                                        for (String tag : upload.keySet()) {
                                            if (replaced == null) {
                                                ((JSONObject)upload.get(tag)).put("timestamp", timestamp);
                                                document.append(tag, upload.get(tag)); continue;
                                            }
                                            Document dim = (Document) replaced.get(tag);
                                            if (dim == null) {
                                                ((JSONObject)upload.get(tag)).put("timestamp", timestamp);
                                                document.append(tag, upload.get(tag));
                                            } else {
                                                Long dimTimestamp = dim.getLong("timestamp");
                                                if (dimTimestamp == null || dimTimestamp < timestamp) {
                                                    ((JSONObject)upload.get(tag)).put("timestamp", timestamp);
                                                    document.append(tag, upload.get(tag));
                                                } else {
                                                    document.append(tag, replaced.get(tag));
                                                }
                                            }
                                        }
                                        return document;
                                    }
                            }, new SimpleQueryFilterCreator().withField("car_vin")), //持久化查询字段
                    persistFields,
                    new TimelyReplacer(), new Fields());


        //登入登出流
        Stream authStream = stage.each(outputFields, new AuthFilter());
    
        MongoMapper authcMapper = new SimpleMongoMapper().withFields
                ("platform_id", "car_vin", "timestamp", "action"); //更新的字段
        MongoState.Options authcOptions = new MongoState.Options()
                .withUrl(url)
                .withDatabase(database)
                .withUsername(username)
                .withPassword(password)
                .withAdmin(admin)
                .withCollectionName(collectionName1)
                .withMapper(authcMapper);
        authStream.partitionPersist(new MongoStateFactory(authcOptions),
                new Fields("platform_id", "car_vin", "timestamp", "action"),
                new MongoStateUpdater(), new Fields() ); //记录登录登出

        Config config = new Config();

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, topology.build());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-realtime", config, topology.build());
        }
    }
}

